package com.neuedu.heart

import org.apache.spark.SparkConf
import org.apache.spark.sql.{SaveMode, SparkSession}

/**
 * 每个年龄段患有心血管疾病的百分比
 */
object AgeInCardioPercent {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME","root")
    //-1. spark 配置对象
    val sparkConf = new SparkConf()
      .setMaster("local[*]")
    //-2. sparksession对象
    val sparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .enableHiveSupport()  //-:开启hive的支持
      .appName("AgeInCardioPercent")
      .getOrCreate()

    // 读取hive表的数据
    // 原始数据集中性别存储的为 1，2
    // if(gender == 2,1,0) as gender，将性别为 2 的进行转换，只保存 0，1
    val df = sparkSession.sql(
      """
        |select if(gender == 2,1,0) as gender,case
        |   when age < 30 then '30以下'
        |   when age < 35 then '30-35'
        |   when age < 40 then '35-40'
        |   when age < 45 then '40-45'
        |   when age < 50 then '45-50'
        |   when age < 55 then '50-55'
        |   when age < 60 then '55-60'
        |   when age < 65 then '60-65'
        |   when age < 70 then '65-70'
        |   when age >= 70 then '70以上'
        |   end level
        |from db_heart.dwd_cardio where cardio = 1
        |""".stripMargin
    )
    // 将dataframe 转换为临时表
    val cardio_view = df.createTempView("cardio_view")

    // 1. 每个年龄段患有心血管疾病的人数
    val dws_cardio_age = sparkSession.sql("select level as name,count(*) as value from cardio_view group by level order by level")
    //dws_cardio_age.show()
    //-将结果进行存储，存储到 hive 数据库中
    dws_cardio_age.write.mode(SaveMode.Overwrite).saveAsTable("db_heart.dws_cardio_age")

    // 2. 每个年龄段男性患有心血管的人数
    val dws_cardio_age_male = sparkSession.sql("select level as name,count(*) as value from cardio_view where gender = 1 group by level order by level")
    //dws_cardio_age_male.show()
    dws_cardio_age_male.write.mode(SaveMode.Overwrite).saveAsTable("db_heart.dws_cardio_age_male")

    // 3. 每个年龄段女性患有心血管的人数
    val dws_cardio_age_female = sparkSession.sql("select level as name,count(*) as value from cardio_view where gender = 0 group by level order by level")
    //dws_cardio_age_female.show()
    dws_cardio_age_female.write.mode(SaveMode.Overwrite).saveAsTable("db_heart.dws_cardio_age_female")

    // sparkSession的停止、关闭
    sparkSession.stop()
    sparkSession.close()
  }
}
